package com.rem.flink.flink10Sql;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * 时间属性和窗口
 *
 * @author Rem
 * @date 2022-11-07
 */

public class TimeAndWindowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 在创建表的DDL中直接定义时间属性
        String createDdl = "CREATE TABLE clickTable (" +
                " user_name STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " et AS TO_TIMESTAMP( FROM_UNIXTIME(ts / 1000) ), " +  //BIGINT类型转为时间戳类型
                " WATERMARK FOR et AS et - INTERVAL '1' SECOND " +    //定义1s延迟时间的水位线
                ") WITH (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'input/clicks.csv', " +
                " 'format' =  'csv' " +
                ")";

        tableEnv.executeSql(createDdl);

        /**
         * 事件时间
         */
        SingleOutputStreamOperator<Event> clickStream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((event, l) -> event.getTimestamp()));

        Schema build = Schema.newBuilder()
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .column("timestamp", DataTypes.BIGINT())
                .columnByExpression("et", "CAST(TO_TIMESTAMP(FROM_UNIXTIME(`timestamp`)) AS TIMESTAMP(3))")
                .watermark("et", "et - INTERVAL '10' SECOND")
                .build();
        Table clickTable = tableEnv.fromDataStream(clickStream, build);
        // clickTable.printSchema();


        // 聚合查询转换
        /**
         * 分组聚合
         */
        Table aggTable = tableEnv.sqlQuery("SELECT user_name, COUNT(1) FROM clickTable GROUP BY user_name");
        tableEnv.toChangelogStream(aggTable).print("aggTable");

        /**
         *  分组窗口聚合   老版本
         * TUMBLE(time_attr, interval)
         * 定义一个滚动窗口，第一个参数是时间字段，第二个参数是窗口长度。
         */
        Table groupWindowResultTable = tableEnv.sqlQuery("SELECT " +
                "user_name, " +
                "COUNT(1) AS cnt, " +
                "TUMBLE_END(et, INTERVAL '10' SECOND) as endT " +
                "FROM clickTable " +
                "GROUP BY " +                     // 使用窗口和用户名进行分组
                "  user_name, " +
                "  TUMBLE(et, INTERVAL '10' SECOND)" // 定义10s滚动窗口
        );
        tableEnv.toChangelogStream(groupWindowResultTable).print("groupWindowResultTable:");

        env.execute();
    }
}
